package com.sanfu.etl.config;

import com.sanfu.etl.pojo.KafkaProps;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FlinkKafkaConsumerConfiguration {
    @Autowired
    KafkaProps props;

    public FlinkKafkaConsumer<String> stringFlinkKafkaConsumerA(){
        KafkaProps.Consumer consumerA = props.getConsumerA();
        return new FlinkKafkaConsumer<>(
                consumerA.getTopic(),
                new SimpleStringSchema(),
                consumerA.kafkaProps()
        );
    }

    public FlinkKafkaConsumer<String> stringFlinkKafkaConsumerB(){
        KafkaProps.Consumer consumerB = props.getConsumerB();
        return new FlinkKafkaConsumer<>(
                consumerB.getTopic(),
                new SimpleStringSchema(),
                consumerB.kafkaProps()
        );
    }
}
